package com.hxh.rpc.provider;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.hxh.rpc.registry.ServiceRegistry;
import com.hxh.rpc.registry.model.ServiceMetaConfig;
import com.hxh.rpc.util.ServiceUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;

/**
 * 服务提供.
 * <p>
 * 负责扫描服务provider注解的bean,注册服务到注册中心，启动netty监听。
 * 提供RPC请求实际处理
 * </p>
 *
 * @author hxh
 * @date 2022/3/2
 * @since 1.0
 */
@Slf4j
public class RpcProviderBean implements InitializingBean, BeanPostProcessor {

    private String address;

    private ServiceRegistry serviceRegistry;

    private Map<String, Object> providerBeanMap = new ConcurrentHashMap<>();

    private EventLoopGroup parentGroup = null;

    private EventLoopGroup workerGroup = null;

    private static ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("rpc-provider-%d").build();

    private static ThreadPoolExecutor rpcThreadPoolExecutor;

    public RpcProviderBean(String address, ServiceRegistry serviceRegistry) {
        this.address = address;
        this.serviceRegistry = serviceRegistry;
    }

    @Override
    public void afterPropertiesSet() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("rpc-netty-%d").build();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024), threadFactory, new ThreadPoolExecutor.AbortPolicy());
        threadPoolExecutor.submit(() -> {
            try {
                startNettyServer();
            } catch (InterruptedException e) {
                log.error("start nettyServer error", e);
            }
        });
    }

    private void startNettyServer() throws InterruptedException {
        if (Objects.nonNull(workerGroup) && Objects.nonNull(parentGroup)) {
            return;
        }
        log.info("start nettySerer begin");
        parentGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(parentGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast(
                                        new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 0),
                                        new ObjectDecoder(ClassResolvers.weakCachingResolver(this.getClass().getClassLoader())),
                                        new ObjectEncoder(),
                                        new RpcProviderNettyHandler(providerBeanMap))
                        ;
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 512)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
        ;
        String[] array = address.split(":");
        String host = array[0];
        int port = Integer.parseInt(array[1]);
        //启动netty服务
        ChannelFuture future = serverBootstrap.bind(host, port).sync();
        log.info("start nettySerer end.host:{}, port:{}", host, port);
        future.channel().closeFuture().sync();
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        //获取注解bean
        RpcProvider rpcProvider = bean.getClass().getAnnotation(RpcProvider.class);
        if (rpcProvider == null) {
            //无注解直接返回
            return bean;
        }
        saveCache(rpcProvider);
        registerProviderService(rpcProvider);
        return bean;
    }

    private void registerProviderService(RpcProvider rpcProvider) {
        String[] addressArray = address.split(ServiceUtils.SPLIT_CHAR);
        String host = addressArray[0];
        int port = Integer.parseInt(addressArray[1]);
        ServiceMetaConfig serviceMetaConfig = ServiceMetaConfig.builder().host(host)
                .port(port)
                .version(rpcProvider.serviceVersion())
                .serviceName(rpcProvider.serviceInterface().getName())
                .build();
        try {
            serviceRegistry.registry(serviceMetaConfig);
            log.info("register service success, serviceMetaConfig: {}", serviceMetaConfig);
        } catch (Exception e) {
            log.error("register service error, serviceMetaConfig: {}", serviceMetaConfig);
            log.error(e.getMessage(), e);
        }
    }

    private void saveCache(RpcProvider rpcProvider) {
        //缓存保存
        String serviceName = rpcProvider.serviceInterface().getName();
        String version = rpcProvider.serviceVersion();
        providerBeanMap.put(ServiceUtils.buildServiceKey(serviceName, version), rpcProvider);
        log.info("find rpcProvider. serviceName:{}, version:{}", serviceName, version);
    }

    /**
     * 提交rpc处理任务
     *
     * @param task
     */
    public static void submit(Runnable task) {
        if (rpcThreadPoolExecutor == null) {
            synchronized (RpcProvider.class) {
                if (rpcThreadPoolExecutor == null) {
                    TimeUnit unit;
                    BlockingQueue workQueue;
                    rpcThreadPoolExecutor = new ThreadPoolExecutor(100,
                            100,
                            600L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1024), threadFactory);
                }
            }
        }
        rpcThreadPoolExecutor.submit(task);
    }
}
